package com.yifeng.repo.controller.excel.loader.task;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.excel.support.ExcelTypeEnum;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.gomcarter.frameworks.base.exception.CustomException;
import com.google.common.base.Strings;
import com.talkyun.utils.Looper;
import com.yifeng.repo.base.utils.file.FileHelper;
import com.yifeng.repo.controller.excel.loader.biz.BizImportAction;
import com.yifeng.repo.controller.excel.loader.dto.DataTaskConf;
import com.yifeng.repo.controller.excel.loader.dto.DataTaskDto;
import com.yifeng.repo.controller.excel.loader.dto.RowDataDto;
import com.yifeng.repo.controller.excel.loader.dto.RowResultDto;
import com.yifeng.repo.controller.excel.loader.redis.ExcelLoaderRedisWorker;
import com.yifeng.repo.controller.excel.loader.task.handler.ExcelFileHandler;
import com.yifeng.repo.controller.excel.loader.task.handler.ExcelWidthStyleStrategy;
import com.yifeng.repo.controller.excel.loader.task.listener.MapDataAsyncReadListener;
import com.yifeng.repo.tookit.oss.OssFileClient;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by daibing on 2023/1/23.
 */
@Slf4j
public class DataImportTask {
    /**
     * 构造函数传入
     */
    private final DataTaskDto dataTaskDto;
    private final BizImportAction bizImportAction;
    private final DataTaskConf properties;
    private final ExcelLoaderRedisWorker redisWorker;
    private final OssFileClient ossFileClient;

    /**
     * 本地文件：导入excel到本地转存文件，以及导入日志转存文件
     */
    private File importDumpFile;
    private File importLogOkDumpFile;
    private File importLogErrorDumpFile;
    private Map<Integer, String> importLogTemplateHeader;
    private ExcelWriter importLogOkDumpFileWriter;
    private ExcelWriter importLogErrorDumpFileWriter;
    private WriteSheet importLogOkDumpFileSheet;
    private WriteSheet importLogErrorDumpFileSheet;

    /**
     * 启动阶段初始化
     */
    private BlockingQueue<RowDataDto> rowDataQueue;
    private Looper reader;
    private Looper writer;
    private Looper checker;

    /**
     * 计算器：导入不需要偏移量（组件自己连续读excel即可）
     */
    private final AtomicInteger readerCount = new AtomicInteger(0);
    private final AtomicInteger writerCount = new AtomicInteger(0);
    private volatile int totalNumber = 0;
    private volatile int finishedNumber = 0;
    private volatile int errorNumber = 0;
    private volatile boolean takeFinished = false;
    private volatile String dataResultUrl = null;

    public DataImportTask(DataTaskDto dataTaskDto, BizImportAction bizImportAction, DataTaskConf properties,
                          ExcelLoaderRedisWorker redisWorker, OssFileClient ossFileClient) {
        this.dataTaskDto = dataTaskDto;
        this.bizImportAction = bizImportAction;
        this.properties = properties;
        this.redisWorker = redisWorker;
        this.ossFileClient = ossFileClient;
    }

    public void startup() {
        this.initDumpFile();
        this.checkAndUpdateTask();
        this.initDumpFileEasyExcelWrite();
        rowDataQueue = new ArrayBlockingQueue<>(properties.getQueueSize());
        reader = this.buildReader(dataTaskDto);
        writer = this.buildWriter(dataTaskDto);
        checker = this.buildChecker(dataTaskDto);
        reader.start();
        writer.start();
        checker.start();
    }

    public void shutdown() {
        reader.shutdown();
        writer.shutdown();
        checker.shutdown();
    }

    public void shutdown4Error(DataTaskDto dataTaskDto) {
        importDumpFile.delete();
        importLogOkDumpFile.delete();
        importLogErrorDumpFile.delete();
        bizImportAction.failed(dataTaskDto);
        shutdown();
    }

    private void initDumpFile() {
        this.importDumpFile = new File(dataTaskDto.getDataDumpFilePath());
        this.importLogOkDumpFile = ExcelFileHandler.getOkLogFile(dataTaskDto.getDataDumpFilePath());
        this.importLogErrorDumpFile = ExcelFileHandler.getErrorLogFile(dataTaskDto.getDataDumpFilePath());
    }

    private void checkAndUpdateTask() {
        // 当前还未开始处理，也就还没有开始偏移
        if (dataTaskDto.getFinishedNumber() == 0) {
            return;
        }
        // 当前已经开始处理了，也就是有偏移了，并且转存文件和日志文件都存在就从偏移量开始处理
        if (this.importDumpFile.exists() && this.importLogOkDumpFile.exists()) {
            return;
        }
        // 如果转存文件不存在，就需要从头开始
        dataTaskDto.setFinishedNumber(0);
        dataTaskDto.setErrorNumber(0);
    }

    private void checkAndDownloadDumpFile() {
        if (!this.importDumpFile.exists()) {
            try {
                // 将导入件拷贝为转存文件
                FileHelper.downloadFile(dataTaskDto.getDataFileUrl(), this.importDumpFile, properties.getImportExcelDownTaskNumber());
            } catch (IOException e) {
                throw new CustomException(dataTaskDto.getRequestKey() + " 根据导入文件初始化转存文件失败: " + dataTaskDto.getDataFileUrl() + ", " + importDumpFile.getPath());
            }
        }
    }

    private void initDumpFileEasyExcelWrite() {
        // 因为下载文件后移到reader线程了，所以初始化的时候还没有下载导入转存文件
        if (!importDumpFile.getParentFile().exists()) {
            importDumpFile.getParentFile().mkdirs();
        }
        this.importLogTemplateHeader = buildImportLogTemplateHeader();
        this.importLogOkDumpFileWriter = EasyExcel.write()
                .registerWriteHandler(new ExcelWidthStyleStrategy(bizImportAction))
                .file(importLogOkDumpFile).excelType(ExcelTypeEnum.XLSX)
                .build();
        if (properties.isImportLogFileSplit()) {
            this.importLogErrorDumpFileWriter = EasyExcel.write()
                    .registerWriteHandler(new ExcelWidthStyleStrategy(bizImportAction))
                    .file(importLogErrorDumpFile).excelType(ExcelTypeEnum.XLSX)
                    .build();
        }
        // 是否需要写头：重启的情况下，即使是第一次也不应该再写头
        if (bizImportAction.headRowNumber() >= dataTaskDto.getFinishedNumber()) {
            this.importLogOkDumpFileSheet = EasyExcel.writerSheet().head(this.buildResultHead(importLogTemplateHeader)).build();
            this.importLogErrorDumpFileSheet = EasyExcel.writerSheet().head(this.buildResultHead(importLogTemplateHeader)).build();
        } else {
            this.importLogOkDumpFileSheet = EasyExcel.writerSheet().needHead(false).build();
            this.importLogErrorDumpFileSheet = EasyExcel.writerSheet().needHead(false).build();
        }
    }

    /**
     * 导入读线程主体业务逻辑
     */
    private Looper buildReader(DataTaskDto dataTaskDto) {
        return new Looper(dataTaskDto.getRequestKey() + "-import-reader", 50, 1000) {
            private final MapDataAsyncReadListener asyncReadListener = this.buildAsyncReadListener(dataTaskDto);
            private final int startRowNumber = Math.max(bizImportAction.headRowNumber(), dataTaskDto.getFinishedNumber());

            @Override
            protected void loop() throws Throwable {
                // 下载并且加载导入转存文件
                checkAndDownloadDumpFile();

                // 开始读取excel
                EasyExcel.read(importDumpFile, this.asyncReadListener)
                        .sheet()
                        .headRowNumber(startRowNumber)
                        .doRead();
                log.info("import task {} reader finished: readerCount={}", dataTaskDto.getRequestKey(), readerCount.get());
                bizImportAction.readFinished(dataTaskDto);
                this.shutdown();
            }

            private MapDataAsyncReadListener buildAsyncReadListener(DataTaskDto dataTaskDto) {
                return new MapDataAsyncReadListener(
                        bizImportAction.templateHeader(),
                        properties.getBatchSize(),
                        rowDataDtoList -> {
                            // 行数据写入内部队列
                            for (RowDataDto rowDataDto : rowDataDtoList) {
                                try {
                                    rowDataQueue.put(rowDataDto);
                                } catch (InterruptedException e) {
                                    throw new CustomException(dataTaskDto.getRequestKey() + " 读取导入文件的数据写入内部队列失败: " + rowDataDto.getRowNumber());
                                }
                                if (rowDataDto.getRowNumber() == Integer.MIN_VALUE) {
                                    break;
                                }
                                readerCount.getAndIncrement();
                            }
                        },
                        totalRowNumber -> {
                            totalNumber = totalRowNumber - bizImportAction.headRowNumber();
                        }
                );
            }

            @Override
            protected void loopThrowable(Throwable t) {
                log.warn("import task {} reader failed: error=", dataTaskDto.getRequestKey(), t);
                dataTaskDto.setErrorMessage("reader failed by " + t.getMessage());
                shutdown4Error(dataTaskDto);
            }
        };
    }

    /**
     * 导入写线程主体业务逻辑
     */
    private Looper buildWriter(DataTaskDto dataTaskDto) {
        return new Looper(dataTaskDto.getRequestKey() + "-import-writer", 50, 1000) {

            @Override
            protected void loop() throws Throwable {
                // 1. 准备一批数据
                List<RowDataDto> rowDataDtoList = new ArrayList<>(properties.getBatchSize());
                for (int i = 0; i < properties.getBatchSize(); i++) {
                    RowDataDto rowDataDto = rowDataQueue.take();
                    // 读取到结束标记位
                    if (rowDataDto.getRowNumber() == Integer.MIN_VALUE) {
                        takeFinished = true;
                        break;
                    }
                    rowDataDtoList.add(rowDataDto);
                    writerCount.getAndIncrement();
                }

                // 2. 调度 BizImportAction#sink 方法推送批次数据，并且将数据导入结果写入日志文件
                if (!rowDataDtoList.isEmpty()) {
                    List<RowResultDto> rowResultDtoList = bizImportAction.sink(rowDataDtoList);
                    List<List<String>> okRowList = this.buildResultData(rowResultDtoList, importLogTemplateHeader, properties.isImportLogFileSplit(), true);
                    importLogOkDumpFileWriter.write(okRowList, importLogOkDumpFileSheet);
                    if (properties.isImportLogFileSplit()) {
                        List<List<String>> errorRowList = this.buildResultData(rowResultDtoList, importLogTemplateHeader, properties.isImportLogFileSplit(), false);
                        importLogErrorDumpFileWriter.write(errorRowList, importLogErrorDumpFileSheet);
                    }
                    finishedNumber = writerCount.get();
                    errorNumber += rowResultDtoList.stream().filter(rowResultDto -> !rowResultDto.isSuccess()).count();
                }

                // 3. 数据处理完成就将转存文件上传到oss
                if (takeFinished) {
                    importDumpFile.delete();
                    importLogOkDumpFileWriter.finish();
                    String importOkLogFileUri = bizImportAction.getOkLogFileUri(dataTaskDto.getRequestKey(), dataTaskDto.getType());
                    dataResultUrl = ossFileClient.uploadFile(importOkLogFileUri, importLogOkDumpFile.getAbsolutePath());
                    importLogOkDumpFile.delete();
                    if (properties.isImportLogFileSplit()) {
                        importLogErrorDumpFileWriter.finish();
                        String importErrorLogFileUri = bizImportAction.getErrorLogFileUri(dataTaskDto.getRequestKey(), dataTaskDto.getType());
                        dataResultUrl += "," + ossFileClient.uploadFile(importErrorLogFileUri, importLogErrorDumpFile.getAbsolutePath());
                        importLogErrorDumpFile.delete();
                    }
                    log.info("import task {} writer finished: readerCount={}, writerCount={}, dataResultUrl={}",
                            dataTaskDto.getRequestKey(), readerCount.get(), writerCount.get(), dataResultUrl);
                    bizImportAction.writeFinished(dataTaskDto);
                    this.shutdown();
                }
            }

            /**
             * 注意内部的List<String>是一行数据，外面的List是多行数据
             */
            private List<List<String>> buildResultData(List<RowResultDto> rowResultDtoList, Map<Integer, String> resultTemplateHeader,
                                                       boolean importLogFileSplit, boolean success) {
                List<List<String>> rowList = new ArrayList<>(rowResultDtoList.size());
                for (RowResultDto rowResultDto : rowResultDtoList) {
                    if (importLogFileSplit && rowResultDto.isSuccess() != success) {
                        continue;
                    }
                    List<String> dataList = new ArrayList<>(resultTemplateHeader.size());
                    // 先拷贝源数据，注意Map数据对象必须是 LinkedHashMap 保持顺序
                    if (rowResultDto.getData() != null && rowResultDto.getData().size() > 0) {
                        dataList.addAll(rowResultDto.getData().values());
                    }
                    // 再拷贝3个固定字段
                    dataList.add(String.valueOf(rowResultDto.getRowNumber()));
                    dataList.add(String.valueOf(rowResultDto.isSuccess()));
                    dataList.add(String.valueOf(rowResultDto.getMessage()));
                    rowList.add(dataList);
                }
                return rowList;
            }

            @Override
            protected void loopThrowable(Throwable t) {
                log.warn("import task {} writer failed: error=", dataTaskDto.getRequestKey(), t);
                dataTaskDto.setErrorMessage("writer failed by " + t.getMessage());
                shutdown4Error(dataTaskDto);
            }
        };
    }

    /**
     * 导入检查线程主体业务逻辑
     */
    private Looper buildChecker(DataTaskDto dataTaskDto) {
        return new Looper(dataTaskDto.getRequestKey() + "-import-checker", 1000, 1000) {
            @Override
            protected void loop() throws Throwable {
                // 1. 更新数据任务对象
                Date now = new Date();
                boolean totalFinished = takeFinished && !Strings.isNullOrEmpty(dataResultUrl);
                dataTaskDto.setDataResultUrl(dataResultUrl);
                dataTaskDto.setTotalFinished(totalFinished);
                dataTaskDto.setTotalNumber(totalFinished ? readerCount.get() : totalNumber);
                dataTaskDto.setFinishedNumber(finishedNumber);
                dataTaskDto.setErrorNumber(errorNumber);
                dataTaskDto.setTotalFinishedTime(totalFinished ? now : null);
                dataTaskDto.setRefreshTime(now);

                // 2. 写入redis，返回写入内容
                String content = redisWorker.saveDataTask(dataTaskDto);

                // 3. 读线程和写线程都处理完成，并且上传文件到oss得到数据结果url
                if (totalFinished) {
                    log.info("import task {} checker finished: content={}", dataTaskDto.getRequestKey(), content);
                    bizImportAction.totalFinished(dataTaskDto);
                    this.shutdown();
                }
            }

            @Override
            protected void loopThrowable(Throwable t) {
                log.warn("import task {} checker failed: error=", dataTaskDto.getRequestKey(), t);
                dataTaskDto.setErrorMessage("checker failed by " + t.getMessage());
                shutdown4Error(dataTaskDto);
            }
        };
    }

    private Map<Integer, String> buildImportLogTemplateHeader() {
        Map<Integer, String> resultTemplateHeader = new LinkedHashMap<>();
        // 导入日志业务头
        int i = 0;
        for (String header : bizImportAction.importLogBizHeader().split(",")) {
            resultTemplateHeader.put(i++, header);
        }
        // 固定的公共头
        for (String header : properties.getImportLogBaseHeader().split(",")) {
            resultTemplateHeader.put(i++, header);
        }
        return resultTemplateHeader;
    }

    /**
     * 注意内部的List<String>是一个头（又可以区分多个小头），外面的List是多个头
     */
    private List<List<String>> buildResultHead(Map<Integer, String> resultTemplateHeader) {
        List<List<String>> head = new ArrayList<>(resultTemplateHeader.size());
        for (int i = 0; i < resultTemplateHeader.size(); i++) {
            List<String> sub = new ArrayList<>(1);
            sub.add(resultTemplateHeader.get(i));
            head.add(sub);
        }
        return head;
    }
}
